from __future__ import annotations

import json
import os
import platform
import shutil
import subprocess
import sys
import typing as _t
from contextlib import contextmanager
from dataclasses import dataclass, field
from functools import partial
from importlib.metadata import version as version_of
from pathlib import Path
from textwrap import dedent

import pytest
import tomli_w
from click.testing import CliRunner
from pip._internal.commands.install import InstallCommand
from pip._internal.index.package_finder import PackageFinder
from pip._internal.models.candidate import InstallationCandidate
from pip._internal.models.link import Link
from pip._internal.network.session import PipSession
from pip._internal.req.constructors import (
    install_req_from_editable,
    install_req_from_line,
)
from pip._internal.utils.direct_url_helpers import direct_url_from_link
from pip._vendor.packaging.version import Version
from pip._vendor.pkg_resources import Requirement

from piptools._compat import Distribution
from piptools.cache import DependencyCache
from piptools.exceptions import NoCandidateFound
from piptools.locations import DEFAULT_CONFIG_FILE_NAMES
from piptools.logging import log
from piptools.repositories import PyPIRepository
from piptools.repositories.base import BaseRepository
from piptools.resolver import BacktrackingResolver, LegacyResolver
from piptools.utils import (
    PIP_VERSION,
    as_tuple,
    is_url_requirement,
    key_from_ireq,
    make_install_requirement,
)

from .constants import MINIMAL_WHEELS_PATH, TEST_DATA_PATH
from .utils import looks_like_ci


@dataclass
class FakeOptions:
    features_enabled: list[str] = field(default_factory=list)
    deprecated_features_enabled: list[str] = field(default_factory=list)
    target_dir: str | None = None


class FakeRepository(BaseRepository):
    def __init__(self, options: FakeOptions):
        self._options = options

        with open(os.path.join(TEST_DATA_PATH, "fake-index.json")) as f:
            self.index = json.load(f)

        with open(os.path.join(TEST_DATA_PATH, "fake-editables.json")) as f:
            self.editables = json.load(f)

    def get_hashes(self, ireq):
        # Some fake hashes
        return {
            "test:123",
            "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
        }

    def find_best_match(self, ireq, prereleases=False):
        if ireq.editable:
            return ireq

        versions = list(
            ireq.specifier.filter(
                self.index[key_from_ireq(ireq)], prereleases=prereleases
            )
        )
        if not versions:
            tried_versions = [
                InstallationCandidate(ireq.name, version, "https://fake.url.foo")
                for version in self.index[key_from_ireq(ireq)]
            ]
            raise NoCandidateFound(ireq, tried_versions, ["https://fake.url.foo"])
        best_version = max(versions, key=Version)
        return make_install_requirement(key_from_ireq(ireq), best_version, ireq)

    def get_dependencies(self, ireq):
        if ireq.editable or is_url_requirement(ireq):
            return self.editables[str(ireq.link)]

        name, version, extras = as_tuple(ireq)
        # Store non-extra dependencies under the empty string
        extras += ("",)
        dependencies = [
            dep for extra in extras for dep in self.index[name][version][extra]
        ]
        return [
            install_req_from_line(dep, constraint=ireq.constraint)
            for dep in dependencies
        ]

    @contextmanager
    def allow_all_wheels(self):
        # No need to do an actual pip.Wheel mock here.
        yield

    @property
    def options(self):
        return self._options

    @property
    def session(self) -> PipSession:
        """Not used"""

    @property
    def finder(self) -> PackageFinder:
        """Not used"""

    @property
    def command(self) -> InstallCommand:
        """Not used"""


def pytest_collection_modifyitems(config, items):
    for item in items:
        # Mark network tests as flaky
        if item.get_closest_marker("network") and looks_like_ci():
            item.add_marker(pytest.mark.flaky(reruns=3, reruns_delay=2))


@pytest.fixture
def fake_dist():
    def _fake_dist(line, deps=None):
        if deps is None:
            deps = []
        req = Requirement.parse(line)
        key = req.name
        if "==" in line:
            version = line.split("==")[1]
        else:
            version = "0+unknown"
        requires = [Requirement.parse(d) for d in deps]
        direct_url = None
        if req.url:
            direct_url = direct_url_from_link(Link(req.url))
        return Distribution(key, version, requires, direct_url)

    return _fake_dist


@pytest.fixture
def repository():
    return FakeRepository(
        options=FakeOptions(deprecated_features_enabled=["legacy-resolver"])
    )


@pytest.fixture
def pypi_repository(tmpdir):
    return PyPIRepository(
        [
            "--index-url",
            PyPIRepository.DEFAULT_INDEX_URL,
            "--use-deprecated",
            "legacy-resolver",
        ],
        cache_dir=(tmpdir / "pypi-repo"),
    )


@pytest.fixture
def depcache(tmpdir):
    return DependencyCache(tmpdir / "dep-cache")


@pytest.fixture
def resolver(depcache, repository):
    # TODO: It'd be nicer if Resolver instance could be set up and then
    #       use .resolve(...) on the specset, instead of passing it to
    #       the constructor like this (it's not reusable)
    return partial(
        LegacyResolver, repository=repository, cache=depcache, existing_constraints={}
    )


@pytest.fixture
def backtracking_resolver(depcache):
    # TODO: It'd be nicer if Resolver instance could be set up and then
    #       use .resolve(...) on the specset, instead of passing it to
    #       the constructor like this (it's not reusable)
    return partial(
        BacktrackingResolver,
        repository=FakeRepository(options=FakeOptions()),
        cache=depcache,
        existing_constraints={},
    )


@pytest.fixture
def base_resolver(depcache):
    return partial(LegacyResolver, cache=depcache, existing_constraints={})


@pytest.fixture
def from_line():
    def _from_line(*args, **kwargs):
        if PIP_VERSION[:2] <= (23, 0):
            hash_options = kwargs.pop("hash_options", {})
            options = kwargs.pop("options", {})
            options["hashes"] = hash_options
            kwargs["options"] = options
        return install_req_from_line(*args, **kwargs)

    return _from_line


@pytest.fixture
def from_editable():
    return install_req_from_editable


@pytest.fixture
def runner():
    if Version(version_of("click")) < Version("8.2"):
        cli_runner = CliRunner(mix_stderr=False)
    else:
        cli_runner = CliRunner()
    with cli_runner.isolated_filesystem():
        yield cli_runner


@pytest.fixture
def tmpdir_cwd(tmpdir):
    with tmpdir.as_cwd():
        yield Path(tmpdir)


@pytest.fixture
def make_pip_conf(tmpdir, monkeypatch):
    created_paths = []

    def _make_pip_conf(content):
        pip_conf_file = "pip.conf" if os.name != "nt" else "pip.ini"
        path = (tmpdir / pip_conf_file).strpath

        with open(path, "w") as f:
            f.write(content)

        monkeypatch.setenv("PIP_CONFIG_FILE", path)

        created_paths.append(path)
        return path

    try:
        yield _make_pip_conf
    finally:
        for path in created_paths:
            os.remove(path)


@pytest.fixture
def pip_conf(make_pip_conf, minimal_wheels_path):
    return make_pip_conf(
        dedent(
            f"""\
            [global]
            no-index = true
            find-links = {minimal_wheels_path.as_posix()}
            """
        )
    )


@pytest.fixture
def pip_with_index_conf(make_pip_conf, minimal_wheels_path):
    return make_pip_conf(
        dedent(
            f"""\
            [global]
            index-url = http://example.com
            find-links = {minimal_wheels_path.as_posix()}
            """
        )
    )


@pytest.fixture(scope="session")
def make_package(tmp_path_factory):
    """
    Make a package from a given name, version and list of required packages.
    """

    def _make_package(
        name,
        version="0.1",
        install_requires=None,
        extras_require=None,
        build_system_requires=None,
    ):
        if install_requires is None:
            install_requires = []

        if extras_require is None:
            extras_require = dict()

        install_requires_str = "[{}]".format(
            ",".join(f"{package!r}" for package in install_requires)
        )

        package_dir = tmp_path_factory.mktemp("packages") / name / version
        package_dir.mkdir(parents=True)

        with (package_dir / "setup.py").open("w") as fp:
            fp.write(
                dedent(
                    f"""\
                    from setuptools import setup
                    setup(
                        name={name!r},
                        version={version!r},
                        author="pip-tools",
                        author_email="pip-tools@localhost",
                        url="https://github.com/jazzband/pip-tools",
                        install_requires={install_requires_str},
                        extras_require={extras_require},
                        py_modules=[{name!r}],
                    )
                    """
                )
            )

        # Create a README to avoid setuptools warnings.
        (package_dir / "README").touch()

        # Create a module to make the package importable.
        (package_dir / name).with_suffix(".py").touch()

        if build_system_requires:
            with (package_dir / "pyproject.toml").open("w") as fp:
                fp.write(
                    dedent(
                        f"""\
                        [build-system]
                        requires = {json.dumps(build_system_requires)}
                        """
                    )
                )

        return package_dir

    return _make_package


@pytest.fixture(scope="session")
def run_setup_file():
    """
    Run a setup.py file from a given package dir.
    """

    def _run_setup_file(package_dir_path, *args):
        setup_file = package_dir_path / "setup.py"
        return subprocess.run(
            [sys.executable, str(setup_file), *args],
            cwd=str(package_dir_path),
            stdout=subprocess.DEVNULL,
            check=True,
        )  # nosec

    return _run_setup_file


@pytest.fixture(scope="session")
def make_wheel(run_setup_file):
    """
    Make a wheel distribution from a given package dir.
    """

    def _make_wheel(package_dir, dist_dir, *args):
        return run_setup_file(
            package_dir, "bdist_wheel", "--dist-dir", str(dist_dir), *args
        )

    return _make_wheel


@pytest.fixture
def make_sdist(run_setup_file):
    """
    Make a source distribution from a given package dir.
    """

    def _make_sdist(package_dir, dist_dir, *args):
        return run_setup_file(package_dir, "sdist", "--dist-dir", str(dist_dir), *args)

    return _make_sdist


@pytest.fixture
def make_module(tmpdir):
    """
    Make a metadata file with the given name and content and a fake module.
    """

    def _make_module(fname, content):
        path = os.path.join(tmpdir, "sample_lib")
        os.mkdir(path)
        path = os.path.join(tmpdir, "sample_lib", "__init__.py")
        with open(path, "w") as stream:
            stream.write("'example module'\n__version__ = '1.2.3'")
        if fname == "setup.cfg":
            path = os.path.join(tmpdir, "pyproject.toml")
            with open(path, "w") as stream:
                stream.write(
                    "\n".join(
                        (
                            "[build-system]",
                            'requires = ["setuptools"]',
                            'build-backend = "setuptools.build_meta"',
                        )
                    )
                )
        path = os.path.join(tmpdir, fname)
        with open(path, "w") as stream:
            stream.write(dedent(content))
        return path

    return _make_module


@pytest.fixture(scope="session")
def fake_dists(tmp_path_factory, make_package, make_wheel):
    """
    Generate distribution packages `small-fake-*`
    """
    dists_path = tmp_path_factory.mktemp("dists")
    pkgs = [
        make_package("small-fake-a", version="0.1"),
        make_package("small-fake-b", version="0.2"),
        make_package("small-fake-c", version="0.3"),
    ]
    for pkg in pkgs:
        make_wheel(pkg, dists_path)
    return dists_path


@pytest.fixture(scope="session")
def fake_dists_with_build_deps(tmp_path_factory, make_package, make_wheel):
    """Generate distribution packages with names that make sense for testing build deps."""
    dists_path = tmp_path_factory.mktemp("dists")
    pkgs = [
        make_package(
            "fake_static_build_dep",
            version="0.1",
            install_requires=["fake_transient_run_dep"],
            build_system_requires=["fake_transient_build_dep"],
        ),
        make_package("fake_dynamic_build_dep_for_all", version="0.2"),
        make_package("fake_dynamic_build_dep_for_sdist", version="0.3"),
        make_package("fake_dynamic_build_dep_for_wheel", version="0.4"),
        make_package("fake_dynamic_build_dep_for_editable", version="0.5"),
        make_package("fake_direct_runtime_dep", version="0.1"),
        make_package("fake_direct_extra_runtime_dep", version="0.2"),
        make_package("fake_transient_build_dep", version="0.3"),
        make_package("fake_transient_run_dep", version="0.3"),
    ]
    for pkg in pkgs:
        make_wheel(pkg, dists_path)
    return dists_path


@pytest.fixture
def venv(tmp_path):
    """Create a temporary venv and get the path of its directory of executables."""
    subprocess.run(
        [sys.executable, "-m", "venv", os.fspath(tmp_path)],
        check=True,
    )
    return tmp_path / ("Scripts" if platform.system() == "Windows" else "bin")


@pytest.fixture(autouse=True)
def _reset_log():
    """
    Since piptools.logging.log is a global variable we have to restore its initial
    state. Some tests can change logger verbosity which might cause a conflict
    with other tests that depend on it.
    """
    log.reset()


@pytest.fixture
def make_config_file(tmpdir_cwd):
    """
    Make a config file for pip-tools with a given parameter set to a specific
    value, returning a ``pathlib.Path`` to the config file.
    """

    def _maker(
        pyproject_param: str,
        new_default: _t.Any,
        config_file_name: str = DEFAULT_CONFIG_FILE_NAMES[0],
        section: str = "pip-tools",
        subsection: str | None = None,
    ) -> Path:
        # Create a nested directory structure if config_file_name includes directories
        config_dir = (tmpdir_cwd / config_file_name).parent
        config_dir.mkdir(exist_ok=True, parents=True)

        # Make a config file with this one config default override
        config_file = tmpdir_cwd / config_file_name

        nested_config = {pyproject_param: new_default}
        if subsection:
            nested_config = {subsection: nested_config}
        config_to_dump = {"tool": {section: nested_config}}
        config_file.write_text(tomli_w.dumps(config_to_dump))
        return _t.cast(Path, config_file.relative_to(tmpdir_cwd))

    return _maker


@pytest.fixture(scope="session")
def setuptools_wheel_path(tmp_path_factory):
    """
    Ensure setuptools is downloaded and return the path to the download directory.
    """
    tmp_wheels_path = tmp_path_factory.mktemp("tmp_wheels")
    subprocess.check_call(
        [
            sys.executable,
            "-Im",
            "pip",
            "download",
            f"--dest={tmp_wheels_path.as_posix()}",
            "--no-deps",
            "setuptools",
        ],
    )
    return tmp_wheels_path


@pytest.fixture(scope="session")
def minimal_wheels_path(setuptools_wheel_path):
    """
    Ensure minimal wheels and setuptools are downloaded and return the path.
    """
    shutil.copytree(MINIMAL_WHEELS_PATH, setuptools_wheel_path, dirs_exist_ok=True)
    return setuptools_wheel_path
